package com.swsc.ai.tasks;

import com.swsc.ai.config.SparkSqlConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @describe: 产品预估购买可能性
 * @author: 容若
 * @created: 2023-12-04 13:09
 */
public class ProdEstimationTask extends ProdEstimationTempTask{
    @Override
    public void createView(SparkSession session, String portraitPath, String prodPath) {
        String sql = "SELECT cust_no FROM t_prod_buy_info WHERE prod_code != 'BF0002' and prod_code != '970161' and prod_code != 'S00001' and prod_code != 'D01007' and prod_code != 'D00007' and prod_code !='D00014' and prod_code != 'D02007' and prod_code != 'F00007' and prod_code != 'F00014' and business_balance >= 3000 ";
        Dataset<Row> prodBuyCust = SparkSqlConf.getDataByCompleteSQL(session, sql);
        prodBuyCust.createOrReplaceTempView("prodBuyCustDF");

        String sql_full = "SELECT distinct(cust_no) FROM t_prod_buy_info";
        Dataset<Row> prodBuyCustFull = SparkSqlConf.getDataByCompleteSQL(session, sql_full);
        prodBuyCustFull.createOrReplaceTempView("prodBuyCustFull");

        Dataset<Row> portraitDF = createRealView(session, portraitPath);
        portraitDF.createOrReplaceTempView("portraitDataDF");
    }

    private Dataset<Row> createRealView(SparkSession session, String hdfsPath) {
        Dataset<Row> wideTableDF = null;
        try {
            wideTableDF = session.read().format("csv").option("header", "true").load(hdfsPath);
        } catch (Exception e) {
            System.err.println("未找到有效的宽表数据，查找路径为：" + hdfsPath);
        }
        return wideTableDF;
    }

    @Override
    public void calculateEstimationResult(SparkSession session, String fileName) {
        /**
         * 树模型正样本
         */
        Dataset<Row> treeModelPositiveSample = session.sql("select 1 as label, u.* from (select distinct cust_no from prodBuyCustDF) as p left join (\n" +
                "    select * from portraitDataDF where normal_status = '正常' and cust_type = '个人' and no_trd_days <= 1100\n" +
                ") as u on p.cust_no = u.client_id");

        /**
         * 树模型负样本
         */
        Dataset<Row> treeModeNegativeSample = session.sql("select 0 as label, u.* from prodBuyCustFull as p right join (\n" +
                "    select * from portraitDataDF where normal_status = '正常' and cust_type = '个人' and no_trd_days <= 1100\n" +
                ") as u on p.cust_no = u.client_id\n" +
                "where p.cust_no is null");

        Dataset<Row> sqlDF = treeModelPositiveSample.unionAll(treeModeNegativeSample);
        // 写入HDFS文件
        sqlDF.coalesce(1).write()
                .format("csv")
                .option("header", "true")
                .option("mode", "overwrite")
                .save(fileName);

    }
}
